package producer

import (
	"errors"
	"github.com/Shopify/sarama"
	"time"
)

var (
	DefVersion sarama.KafkaVersion = sarama.V0_10_2_2 //版本
)

type (
	HandleErrorFunc   func(error)
	HandleSucceedFunc func(*sarama.ProducerMessage)
)

type Producer struct {
	config   *sarama.Config
	producer sarama.AsyncProducer //异步生产者
	e        HandleErrorFunc      //发送失败处理
	s        HandleSucceedFunc    //发送成功处理
}

// NewProducer new a producer.
func NewProducer(brokers []string) (*Producer, error) {
	cfg := sarama.NewConfig()
	cfg.Net.KeepAlive = 60 * time.Second
	cfg.Producer.Return.Successes = true
	cfg.Producer.Return.Errors = true
	cfg.Version = sarama.V0_10_0_1
	cfg.Producer.Flush.Frequency = time.Second
	cfg.Producer.Flush.MaxMessages = 10
	return newProducerWithCfg(brokers, cfg)
}

func newProducerWithCfg(brokers []string, cfg *sarama.Config) (*Producer, error) {
	producer, err := sarama.NewAsyncProducer(brokers, cfg)
	if err != nil {
		return nil, err
	}
	p := new(Producer)
	p.producer = producer
	p.config = cfg

	// 处理push成功或失败后
	go p.handle()
	return p, nil
}

func (p *Producer) Push(topic, blockKey string, msg []byte) error {
	if topic == "" {
		errors.New("topic must not be empty")
	}

	if len(msg) == 0 {
		errors.New("msg must not be empty")
	}

	message := &sarama.ProducerMessage{
		Topic: topic,
		Key:   sarama.StringEncoder(blockKey),
		Value: sarama.ByteEncoder(msg),
	}

	// 发送消息
	p.producer.Input() <- message
	return nil
}

func (p *Producer) handle() {
	for {
		select {
		case err, ok := <-p.producer.Errors():
			{
				if ok {
					if p.e != nil {
						p.e(err)
					}
				}
			}
		case msg, ok := <-p.producer.Successes():
			{
				if ok {
					if p.s != nil {
						p.s(msg)
					}
				}
			}
		}
	}
}

// Close  closes the producer
func (p *Producer) Close() error {
	return p.producer.Close()
}

// HandleError set error handler
func (p *Producer) HandleError(e HandleErrorFunc) {
	p.e = e
}

// HandleSucceed set success handler
func (p *Producer) HandleSucceed(s HandleSucceedFunc) {
	p.s = s
}
